package com.chzc.server;

import com.chzc.entity.DevilMessage;
import com.chzc.initializer.DevilConsumerInitializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.nio.charset.StandardCharsets;


/**
 * 消息发送者
 */
public class DevilSubscribe {

    /**
     * broker地址
     */
    private final String host;

    /**
     * broker端口号
     */
    private final int port;

    /**
     * 监听的队列名称
     */
    private final String queueName;

    /**
     * 每次接收多少消息
     */
    private final Integer num;

    /**
     * 多少秒发送一次请求
     */
    private final Long time;


    private Bootstrap bootstrap;


    /**
     * @param host      ip地址
     * @param port      端口号
     * @param queueName 队列名称
     */
    public DevilSubscribe(String host, int port, String queueName) {
        this.host = host;
        this.port = port;
        this.queueName = queueName;
        this.num = 10000;
        // 默认1000毫秒，也就是1S
        this.time = 1000L;
    }

    /**
     * @param host      ip地址
     * @param port      端口号
     * @param queueName 队列名称
     * @param num       每次接收多少参数
     * @param time      间隔多少毫秒发送请求
     */
    public DevilSubscribe(String host, int port, String queueName, Integer num, Long time) {
        this.host = host;
        this.port = port;
        this.queueName = queueName;
        this.num = num;
        this.time = time;
    }

    private ChannelFuture channelFuture;

    private EventLoopGroup group;

    /**
     * 建立连接
     */
    public void start(SimpleChannelInboundHandler<DevilMessage> channelInboundHandler) {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.group(group);
        //自定义一个初始化类
        bootstrap.handler(new DevilConsumerInitializer(channelInboundHandler));
        try {
            this.channelFuture = bootstrap.connect(this.host, this.port).sync();

            // 循环发送请求
            while (true) {
                DevilMessage devilMessage = new DevilMessage();
                // 内容存放每次接收多少参数
                devilMessage.setLen(num.toString().getBytes(StandardCharsets.UTF_8).length);
                devilMessage.setContent(num.toString().getBytes(StandardCharsets.UTF_8));
                // 间隔毫秒
                devilMessage.setMessageId(this.time.toString().getBytes(StandardCharsets.UTF_8));
                devilMessage.setMessageIdLen(this.time.toString().getBytes(StandardCharsets.UTF_8).length);
                // 队里名称
                devilMessage.setQueueName(queueName.getBytes(StandardCharsets.UTF_8));
                devilMessage.setQueueLen(queueName.getBytes(StandardCharsets.UTF_8).length);
                // 标明是Subscribe类型
                devilMessage.setStatus(3);
                // 说明不是心跳类型
                devilMessage.setPingType(false);
                System.out.println(devilMessage);
                channelFuture.channel().writeAndFlush(devilMessage);
                // 睡眠
                Thread.sleep(this.time);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            group.shutdownGracefully();
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * 释放/关闭连接
     */
    public void release() {
        this.group.shutdownGracefully();
    }

    /**
     * 发送消息
     *
     * @param queueName
     * @param message
     */
    public void sendMag(String queueName, String message) {
        // 消息内容
        byte[] content = message.getBytes(StandardCharsets.UTF_8);
        // 消息长度
        int contentLen = message.getBytes(StandardCharsets.UTF_8).length;
        // 队列名称
        byte[] queue = queueName.getBytes(StandardCharsets.UTF_8);
        // 队列名称长度
        int queueLen = queueName.getBytes(StandardCharsets.UTF_8).length;

        DevilMessage devilMessage = new DevilMessage();
        devilMessage.setLen(contentLen);
        devilMessage.setContent(content);
        devilMessage.setMessageId("o".getBytes(StandardCharsets.UTF_8));
        devilMessage.setMessageIdLen("o".getBytes(StandardCharsets.UTF_8).length);
        devilMessage.setQueueName(queue);
        devilMessage.setQueueLen(queueLen);
        devilMessage.setStatus(1);
        devilMessage.setPingType(false);

        this.channelFuture.channel().writeAndFlush(devilMessage);
    }

    public static void main(String[] args) throws Exception {
//        new DevilClien("127.0.0.1", 7000).start();
    }
}
